package net.wicp.tams.common.binlog.self.sender;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import org.apache.commons.collections.MapUtils;
import org.slf4j.Logger;

import net.wicp.tams.common.Result;
import net.wicp.tams.common.apiext.StringUtil;
import net.wicp.tams.common.binlog.self.bean.SendBean;
import net.wicp.tams.common.binlog.self.bean.SendMoniter;
import net.wicp.tams.common.binlog.self.bean.SendRow;
import net.wicp.tams.common.binlog.self.constant.ParamsVar;
import net.wicp.tams.common.constant.dic.YesOrNo;

public abstract class AbsSender implements ISender {

	protected Logger log;
	protected String taskId;
	protected boolean serialize;// 是否启用proto进行序列化
	protected YesOrNo multiThread = YesOrNo.no;// 是否启用多线程
	protected int threadNum = -1;
	protected int querySize = -1;
	protected int tableNum = -1;// 实例拥有的表数
	protected String splitKey;// 分库分表键
	protected YesOrNo sendByTable = YesOrNo.yes;// 是否按分表规则发送

	protected volatile int insertNum = 0;
	protected volatile int updateNum = 0;
	protected volatile int deleteNum = 0;
	protected volatile long sendTimes = 0;// 发送用时

	@Override
	public final Result sendMsg(SendBean sendBean) {
		long begin = System.currentTimeMillis();
		Result ret = null;
		if (sendByTable == YesOrNo.yes) {
			ret = sendMsgSub(sendBean);
		} else {
			List<SendBean> list = spilSendBean(sendBean);
			for (SendBean sendBean2 : list) {
				sendMsgSub(sendBean2);
			}
			ret = Result.getSuc();
		}
		if (ret.isSuc()) {
			int inccount = sendBean.getRows().size();
			switch (sendBean.getOptType()) {
			case insert:
				// insertNum++;
				insertNum += inccount;
				break;
			case update:
				updateNum += inccount;
				break;
			case delete:
				deleteNum += inccount;
				break;
			default:
				break;
			}
		}
		sendTimes += System.currentTimeMillis() - begin;// 发送用时
		return ret;
	}

	@Override
	public final void init(Logger log, String taskId, Map<String, String> params) {
		this.taskId = taskId;
		this.log = log;
		if (MapUtils.isNotEmpty(params)) {
			if ("true".equals(params.get(ParamsVar.SERIALIZE))) {
				serialize = true;
			}
			if ("yes".equals(params.get(ParamsVar.MULTITHREAD))) {
				multiThread = YesOrNo.yes;
			}
			if (params.containsKey(ParamsVar.THREADNUM)) {
				threadNum = Integer.parseInt(params.get(ParamsVar.THREADNUM));
			}
			if (multiThread == YesOrNo.yes && threadNum <= 0) {
				throw new IllegalArgumentException("多线程需要设置线程数");
			}
			if (multiThread == YesOrNo.no) {
				threadNum = 1;
			}

			if (params.containsKey(ParamsVar.QUERYSIZE)) {
				querySize = Integer.parseInt(params.get(ParamsVar.QUERYSIZE));
			}
			if (params.containsKey(ParamsVar.SPLITKEY)) {
				splitKey = params.get(ParamsVar.SPLITKEY);
			}
			splitKey = StringUtil.hasNull(splitKey, "id");// 没有设置分库分表键就用id

			if (params.containsKey(ParamsVar.TABLENUM)) {
				tableNum = Integer.parseInt(params.get(ParamsVar.TABLENUM));
			}

			if ("yes".equals(params.get(ParamsVar.SENDBYTABLE))) {
				sendByTable = YesOrNo.yes;
			} else if ("no".equals(params.get(ParamsVar.SENDBYTABLE))) {
				sendByTable = YesOrNo.no;
			} else {
				sendByTable = null;
			}

			if (sendByTable == null) {
				// && (tableNum % threadNum == 0 || tableNum % threadNum >=
				// threadNum / 2)表大于线程数，且表均分后的的个数大于线程数的一半
				if (tableNum > 0 && threadNum > 0 && tableNum >= threadNum) {
					sendByTable = YesOrNo.yes;
				} else {
					sendByTable = YesOrNo.no;
				}
			}

		}
		initSub(params);
	}

	/***
	 * 得到监控对象
	 */
	@Override
	public SendMoniter getMoniter() {
		return SendMoniter.builder().insertNum(insertNum).updateNum(updateNum).deleteNum(deleteNum).sendTimes(sendTimes)
				.sendByTable(sendByTable).build();
	}

	@Override
	public void initMoniter(SendMoniter sendMoniter) {
		this.insertNum = sendMoniter.getInsertNum();
		this.updateNum = sendMoniter.getUpdateNum();
		this.deleteNum = sendMoniter.getDeleteNum();
		this.sendTimes = sendMoniter.getSendTimes();
	}

	/**
	 * 得到grid值 4f6e51f7-fbd1-11e5-b795-008cfae412f8:1-32403796 得到 32403796
	 * 
	 * @param sendBean
	 * @return
	 */
	protected long getGtidOrder(SendBean sendBean) {
		int begin = sendBean.getGtid().indexOf("-");
		int end = sendBean.getGtid().indexOf(",");
		String retstr = end > 0 ? sendBean.getGtid().substring(begin + 1, end)
				: sendBean.getGtid().substring(begin + 1);
		return Long.valueOf(retstr);
	}

	protected List<SendBean> spilSendBean(SendBean sendBean) {
		List<SendBean> retlist = new ArrayList<>();
		if (sendBean.getRows().size() == 1) {
			retlist.add(sendBean);
		} else {
			for (SendRow i : sendBean.getRows()) {
				SendBean tempSendBean = SendBean.builder().columnCount(sendBean.getColumnCount())
						.gtid(sendBean.getGtid()).optType(sendBean.getOptType())
						.tableMapBean(sendBean.getTableMapBean()).build();
				tempSendBean.setRows(Arrays.asList(new SendRow[] { i }));
				retlist.add(tempSendBean);
			}
		}
		return retlist;
	}

	public abstract void initSub(Map<String, String> params);

	public abstract Result sendMsgSub(SendBean sendBean);
}
